package com.csdj.middleware;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.springframework.validation.annotation.Validated;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Service
public class ZookeeperService {

    private volatile List<String> name;

    private Logger log = LoggerFactory.getLogger("zookeeper");

    @Value("${zookeeper.host}")
    private String zooServer;

    private CuratorFramework client = null;

    //安装锁
    InterProcessMutex lock = null;


    @PostConstruct
    public void init(){
        if(!StringUtils.hasText(zooServer)){
            return;
        }
        client = CuratorFrameworkFactory.newClient(zooServer,
                new ExponentialBackoffRetry(3000,3));
        client.start();//开始连接zookeeper 直接连接成功会重试3次后，
        //创建锁
        lock = new InterProcessMutex(getZookeeperClient(),"/zoo/lock");
        try {
            addWatcher("/zoo/watcher");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建zookeeper 客户端
     * @return
     */
    public CuratorFramework getZookeeperClient(){
        return client;
    }

    /**
     *
     * @param path 节点路径，
     * @param data 节点上的数据
     * @throws Exception
     */
    public void createNode(String path,String data) throws Exception {
        String res = getZookeeperClient()
                .create()
                .creatingParentsIfNeeded() //如果父节点不存在，自动创建
                .forPath(path,data.getBytes());
        System.out.println(res);
    }

    /**
     * 删除节点
     * @param path
     * @throws Exception
     */
    public void deleteNode(String path) throws Exception {
        getZookeeperClient().delete().deletingChildrenIfNeeded().forPath(path);
    }


    /**
     * 创建临时节点
     * @param path
     * @param data
     * @throws Exception
     */
    public void createTempNode(String path,String data) throws Exception {
        String res = getZookeeperClient().create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL)
                .forPath(path,data.getBytes());
        System.out.println(res);
    }


    /**
     * zookeeper原生监听器，只能执行一次，如需要重复监听， 则需要重复添加
     * @param path
     * @throws Exception
     */
    public void addWatcher(String path) throws Exception {
        //先判断节是否存在，如果没有则创建节点
        if(getZookeeperClient().checkExists().forPath(path) == null){
            getZookeeperClient().create().creatingParentsIfNeeded().forPath(path);
        }
        getZookeeperClient().getData()
                .usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    byte[] data = getZookeeperClient().getData().forPath(path);
                    log.info("节点：{} 发生了：{} 最新数据：{}",watchedEvent.getPath(),watchedEvent.getType(),new String(data));
                    addWatcher(path);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).forPath(path);
    }
    



    /**缓存监听器， 可以重复监听
     * Path Cache : 监控一个ZNode的子节点. 当一个子节点增加， 更新，删除时， Path Cache会改变它的状态， 会包含最新的子节点， 子节点的数据和状态
     * Node Cache : 只是监听某一个特定的节点
     * Tree Cache : 可以监控整个树上的所有节点，类似于PathCache和NodeCache的组合
     * @param path
     * @throws Exception
     */
    public void  addCacheWatcher(String path) throws Exception {
        //先判断节是否存在，如果没有则创建节点
        if(getZookeeperClient().checkExists().forPath(path) == null){
            getZookeeperClient().create().forPath(path);
        }
        final PathChildrenCache pathCache = new PathChildrenCache(getZookeeperClient(), path,true);
        pathCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
                    log.info("添加子节点：{},最新数据：{}",event.getData().getPath(),new String(event.getData().getData()));
                }
                if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                    log.info("删除子节点：{},最新数据：{}",event.getData().getPath(),new String(event.getData().getData()));
                }
                if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                    log.info("修改子节点：{},最新数据：{}",event.getData().getPath(),new String(event.getData().getData()));
                }
            }
        });

        pathCache.start();
    }

    /**
     * 利用zookeeper的强一致性，每个线程都可以创建一个不重复的序列号节点
     * 获取下单资格
     * @param path  节点路径
     * @param total  秒杀数量
     * @return
     */
    public boolean getOrderPermit(String path,int total) throws Exception {
        final String pathSerail = getZookeeperClient().create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT_SEQUENTIAL)
                .forPath(path);
        //截取序列号
        int serialNo = Integer.parseInt(pathSerail.substring(path.length()));
        //判断序列号是否小于总数量，如果小于总数量则有资格下单 否则就不能下单
        if(serialNo>=total){
            //已经卖完了，不能再下单
            return false;
        }
        return true;
    }

    public void lockTest(){
        try {
            lock.acquire();//加锁
            System.out.println("当前线程名："+Thread.currentThread().getName());
            System.out.println("开始执行相关操作");
            TimeUnit.SECONDS.sleep(10);
            System.out.println("操作完成");
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            try {
                lock.release();//释放锁
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }




    public static void main(String[] args) {
        final ZookeeperService service = new ZookeeperService();
        try {
            service.createTempNode("/java/zoo/temp","from java code");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
